package com.shujia.streeam

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext
import org.apache.spark.streaming.{Duration, Durations, StreamingContext}

object Demo7Window {
  def main(args: Array[String]): Unit = {
    /**
      * 统计最近15s单词的数量  每隔10s统计一次
      *
      */


    val conf = new SparkConf().setMaster("local[2]").setAppName("Demo1WordCount")
    conf.set("spark.sql.shuffle.partitions", "2")
    val sc = new SparkContext(conf)


    val sqlContext = new SQLContext(sc)

    import sqlContext.implicits._

    /**
      *
      * 创建 streamming上下文对象  需要指定间隔时间   延迟时间
      *
      */
    val ssc = new StreamingContext(sc, Durations.seconds(5))

    ssc.checkpoint("spark/data/window")

    /**
      * 读取socket创建DS
      * nc -lk 8888
      *
      */

    val studentDS = ssc.socketTextStream("node1", 8888)

    //聚合函数
    val fun = (x: Int, y: Int) => {
      x + y
    }

    /**
      * reduceByKeyAndWindow  窗口操作
      * 需要指定聚合函数   窗口大小  滑动时间
      *
      */

    val kvRDD = studentDS
      .flatMap(_.split(","))
      .map(word => (word, 1))

    ///未优化版本 会有重复计算问题
    kvRDD.reduceByKeyAndWindow(fun, Durations.seconds(15), Durations.seconds(10))
    //.print()

    //j减去多余部分
    val inv = (x: Int, y: Int) => {
      x - y
    }


    //优化之后的版本
    /**
      * 上一个窗口的结果减去多余batch的结果，加上新batch的结果
      *
      */
    kvRDD
      .reduceByKeyAndWindow(fun, inv, Durations.seconds(15), Durations.seconds(10))
      .filter(kv => kv._2 != 0)//过滤掉为0的数据
      .print()


    ssc.start()
    ssc.awaitTermination()
    ssc.stop()


  }
}
